package vip.meeet.kafka.consumer;

import cn.freemethod.kafka.listener.ReceiveListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Subscriber {
	
	private static final Logger logger = LoggerFactory.getLogger(Subscriber.class);

    private Subscriber(){
    	throw new IllegalAccessError("禁止实例化！");
    }

    public static String subscribe(ReceiveListener listener, String topicName,  String cumsumerGroupId, int consumeThreadCount){
        try {
            if(topicName==null){
                throw new IllegalArgumentException("topicName can not be null");
            }
            if(cumsumerGroupId==null || "".equals(cumsumerGroupId.trim())){
                throw new IllegalArgumentException("cumsumerGroupId can not be empty");
            }
            ExecutorService executor=Executors.newFixedThreadPool(consumeThreadCount);
            for(int i=0; i < consumeThreadCount; i++) {
                final String threadName = String.format("[%s]consume thread-%s", topicName, i);
                executor.execute(new MessageConsumer(threadName, cumsumerGroupId, listener, topicName));
            }
            executor.shutdown();
            return "OK";
        } catch(Exception e) {
        	logger.error(e.getMessage());
            return "ERROR";
        }
    }
}
